package com.flink.split_stream.connect;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Objects;

/**
 * 描述:
 * 多流转换连接 TODO 模拟订单实时对账 实时计算第三方支付和订单状态
 *
 * @author yanzhengwu
 * @create 2022-08-23 19:43
 */
public class StreamsConnectConvert {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Tuple3<String, String, Long>> appStream = env.fromElements(
                Tuple3.of("order-1", "app", 1000L),
                Tuple3.of("order-2", "app", 2000L)

        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                    @Override
                    public long extractTimestamp(Tuple3<String, String, Long> element, long recordTimestamp) {
                        return element.f2;
                    }
                }));

        DataStream<Tuple4<String, String, String, Long>> webStream = env.fromElements(
                Tuple4.of("order-1", "web", "", 1000L),
                Tuple4.of("order-2", "web", "", 2000L),
                Tuple4.of("order-3", "web", "", 3000L)

        ).assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple4<String, String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple4<String, String, String, Long>>() {
                    @Override
                    public long extractTimestamp(Tuple4<String, String, String, Long> element, long recordTimestamp) {
                        return element.f3;
                    }
                }));
        //第一种分组连接写法
        appStream.keyBy(app -> app.f0)
                .connect(webStream.keyBy(web -> web.f0));
        //第二种分组连接写法
        appStream.connect(webStream)
                .keyBy(app -> app.f0, web -> web.f0)
                .process(new OrderMathResult())
                .print();

        env.execute();

    }

    /**
     * 具体处理逻辑
     */
    public static class OrderMathResult extends CoProcessFunction<Tuple3<String, String, Long>, Tuple4<String, String, String, Long>, String> {
        //定义2个 状态,互相等待对方是否有数据流过来；
        //先来的先加状态 后来的判断先来的是否有值，如果没有值把自己的状态加进去，后来的判断先来的是否有值，如果先来的值存在则对账成功，否则等5秒定时器过后没有值进来则对账失败
        ValueState<Tuple3<String, String, Long>> appStatus;
        ValueState<Tuple4<String, String, String, Long>> webStatus;


        //这是第一条流 为app端消息
        @Override
        public void processElement1(Tuple3<String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
            //如果非空则对账成功
            if (Objects.nonNull(webStatus.value())) {
                out.collect("对账成功 ：" + value + "  " + webStatus.value());
                webStatus.clear();
            } else {
                //如果是空将当前状态更新进去
                appStatus.update(value);
                //并注册一个定时器等 5秒 另一条数据
                ctx.timerService().registerEventTimeTimer(value.f2 + 5000L);
            }

        }
        //这是第二条流 为web端消息
        @Override
        public void processElement2(Tuple4<String, String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
            //如果非空则对账成功
            if (Objects.nonNull(appStatus.value())) {
                out.collect("对账成功 ：" + appStatus.value() + "  " + value);
                appStatus.clear();
            } else {
                //如果是空将当前状态更新进去
                webStatus.update(value);
                //并注册一个定时器等 5秒 另一条数据
                ctx.timerService().registerEventTimeTimer(value.f3 + 5000L);
            }
        }

        /**
         * TODO 注册定时器必须  重写 onTimer方法 编写具体处理逻辑
         * @param timestamp
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            if (Objects.nonNull(appStatus.value())) {

                out.collect("对账失败 ：" + appStatus.value() + " 第三方支付平台信息未到");

            }
            if (Objects.nonNull(webStatus.value())) {

                out.collect("对账失败 ：" + webStatus.value() + " 第三方支付平台信息未到");
            }

        }

        /**
         * 状态编程必须从 声明周期中读取状态
         * @param parameters
         * @throws Exception
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            appStatus = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple3<String, String, Long>>("appStatus", Types.TUPLE(Types.STRING, Types.STRING, Types.LONG)));

            webStatus = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple4<String, String, String, Long>>("webStatus", Types.TUPLE(Types.STRING, Types.STRING, Types.STRING, Types.LONG)));
        }
    }
}
